package org.apache.hadoop.mapreduce;

import java.io.IOException;

import mapreduce4j.KeyValuesIterator;
import mapreduce4j.ReducerContext;

import org.apache.hadoop.io.WritableComparable;

/** 
 * The base Reducer class which does an Identity function by default.
 */
public abstract class Reducer<KEYIN extends WritableComparable,VALUEIN extends WritableComparable,KEYOUT extends WritableComparable,VALUEOUT extends WritableComparable> {

  public class Context extends ReducerContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {
	public Context(KeyValuesIterator<KEYIN, VALUEIN> input) {
		super(input);
	}
  }

  /**
   * Called once at the start of the task.
   */
  protected void setup(Context context) throws IOException, InterruptedException {
  }

  /**
   * This method is called once for each key. Most applications will define
   * their reduce class by overriding this method. The default implementation
   * is an identity function.
   */
  @SuppressWarnings("unchecked")
  protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context) throws IOException, InterruptedException {
    for(VALUEIN value: values) {
      context.write((KEYOUT) key, (VALUEOUT) value);
    }
  }

  /**
   * Called once at the end of the task.
   */
  protected void cleanup(Context context) throws IOException, InterruptedException {
  }

  /**
   * Runs the reduce over the keys
   */
  public void run(Context context) throws IOException, InterruptedException {
    setup(context);
    while (context.nextKey()) {
      reduce(context.getCurrentKey(), context.getValues(), context);
    }
    cleanup(context);
  }
}